package com.atguigu.day07;

import com.atguigu.bean.Event;
import com.atguigu.day03.Flink05_Source_Custom;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Flink02_SplitStreamByFilter {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //2.获取数据
        DataStreamSource<Event> streamSource = env.addSource(new Flink05_Source_Custom.ClickSource());

        //3.将一条流根据不同的用户拆分成多条流
        SingleOutputStreamOperator<Event> songsongStream = streamSource.filter(new FilterFunction<Event>() {
            @Override
            public boolean filter(Event value) throws Exception {
                return "Songsong".equals(value.user);
            }
        });

        SingleOutputStreamOperator<Event> bingbingStream = streamSource.filter(new FilterFunction<Event>() {
            @Override
            public boolean filter(Event value) throws Exception {
                return "Bingbing".equals(value.user);
            }
        });

        songsongStream.print("Songsong");
        bingbingStream.print("Bingbing");

        env.execute();
    }
}
